/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal/allocator.h"
#include "dal/config_loader.h"
#include "dal_cyclonedds/converter/StatusConverter.hpp"
#include "dal_cyclonedds/converter/StatusMaskConverter.hpp"
#include "dal_cyclonedds/converter/RetTypeConverter.hpp"
#include "dal_cyclonedds/converter/QosConverter.hpp"
#include "dal_cyclonedds/converter/RetTypeConverter.hpp"
#include "dal_cyclonedds/converter/SampleInfoConverter.hpp"
#include "dal_cyclonedds/converter/GuidConverter.hpp"
#include "dal_cyclonedds/listener_impl/SubscriberListenerImpl.hpp"
#include "dal_cyclonedds/listener_impl/DataReaderListenerImpl.hpp"
#include "dal_cyclonedds/utils/log.hpp"
#include "dal_cyclonedds/utils/EntityGetter.hpp"
#include "dal_cyclonedds/utils/TopicSertypeManager.hpp"

// SUBSCRIBER
dal_subscriber_listener_t*
dal_create_subscriber_listener(dal_subscriber_listener_callbacks_t callbacks)
{
    dal_subscriber_listener_t* listener = dal_subscriber_listener_allocate();
    if (listener == nullptr) {
        return nullptr;
    }
    listener->handle_ = new SubscriberListenerImpl(callbacks);
    if (!listener->handle_) {
        LOG_ERROR() << "SubscriberListenerImpl memory allocation failed." << LOG_ENDL;
        dal_subscriber_listener_free(listener);
        return nullptr;
    }

    return listener;
}

dal_ret_t dal_delete_subscriber_listener(dal_subscriber_listener_t* listener)
{
    if (!listener || !listener->handle_) {
        LOG_ERROR_WITH_LOCATION() << "listener is invalid." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    SubscriberListenerImpl* listener_impl =
        reinterpret_cast<SubscriberListenerImpl*>(listener->handle_);
    if (listener_impl->ref_count_.Get() > 0) {
        LOG_ERROR() << "can't delete subscriber listener because it's entity still exist."
                    << LOG_ENDL;
        return DAL_RETCODE_PRECONDITION_NOT_MET;
    }

    delete listener_impl;
    dal_subscriber_listener_free(listener);
    reinterpret_cast<dal_subscriber_listener_t*>(listener)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

dal_subscriber_t* dal_domain_participant_create_subscriber(dal_domain_participant_t* participant,
                                                           dal_subscriber_qos_t* qos,
                                                           dal_subscriber_listener_t* listener,
                                                           dal_status_mask_t mask)
{
    dal_ret_t ret = DAL_RETCODE_OK;

    dds_entity_t c_participant = get_cyclone_domain_participant(participant);
    if (c_participant <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone domain participant failed." << LOG_ENDL;
        return nullptr;
    }

    dal_subscriber_t* d_sub = dal_subscriber_allocate();
    if (!d_sub) {
        LOG_ERROR_WITH_LOCATION() << "dal subscriber allocate failed." << LOG_ENDL;
        return nullptr;
    }

    SubscriberListenerImpl* c_listener_impl =
        listener ? reinterpret_cast<SubscriberListenerImpl*>(listener->handle_) : nullptr;
    dds_listener_t* c_listener = NULL;
    if (c_listener_impl != nullptr) {
        c_listener = c_listener_impl->CycloneDDSListener();
    }

    dds_qos_t* c_qos = dds_create_qos();
    ret = QosConverter::GetInstance()->ConvertSubscriberQosFromDalToCyclonedds(qos, c_qos);
    if (DAL_RETCODE_OK != ret) {
        LOG_ERROR_WITH_LOCATION() << "convert subscriber qos failed." << LOG_ENDL;
        dal_subscriber_free(d_sub);
        dds_delete_qos(c_qos);
        return nullptr;
    }

    dds_entity_t c_sub = dds_create_subscriber(c_participant, c_qos, c_listener);
    dds_delete_qos(c_qos);
    if (c_sub < 0) {
        LOG_ERROR_WITH_LOCATION() << "create subscriber failed." << LOG_ENDL;
        dal_subscriber_free(d_sub);
        return nullptr;
    }

    d_sub->handle_ = new DalSubscriberImpl(c_sub, c_listener_impl);

    uint32_t c_mask = 0;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToCyclonedds(mask, &c_mask);
    StatusMaskConverter::GetInstance()->FilterSubscriberStatusMask(&c_mask);
    if (dds_set_status_mask(c_sub, c_mask) != DDS_RETCODE_OK) {
        LOG_ERROR_WITH_LOCATION() << "set subscriber status mask failed." << LOG_ENDL;
        return d_sub;
    }

    TopicSertypeManager::GetInstance()->AddEntityToParticipant(c_participant, c_sub);

    return d_sub;
}

dal_subscriber_t*
dal_domain_participant_create_subscriber_with_profile(dal_domain_participant_t* participant,
                                                      dal_subscriber_qos_t* qos,
                                                      const char* profile_name,
                                                      dal_subscriber_listener_t* listener,
                                                      dal_status_mask_t mask)
{
    return dal_domain_participant_create_subscriber(participant, qos, listener, mask);
}

dal_ret_t dal_domain_participant_delete_subscriber(dal_domain_participant_t* participant,
                                                   dal_subscriber_t* sub)
{
    dds_return_t c_ret;
    dds_entity_t c_participant = get_cyclone_domain_participant(participant);
    if (c_participant <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone domain participant failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    dds_entity_t c_sub = get_cyclone_subscriber(sub);
    if (c_sub <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone subscriber failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (c_participant != dds_get_parent(c_sub)) {
        LOG_ERROR_WITH_LOCATION() << "subscriber is not belong to participant." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    const int MAX_CHILDREN_NUM = 100;
    dds_entity_t* children = (dds_entity_t*)malloc(sizeof(dds_entity_t) * MAX_CHILDREN_NUM);
    c_ret = dds_get_children(c_sub, children, MAX_CHILDREN_NUM);
    if (c_ret > 0) {
        LOG_ERROR_WITH_LOCATION() << "subscriber still has children." << LOG_ENDL;
        free(children);
        return DAL_RETCODE_PRECONDITION_NOT_MET;
    }
    free(children);

    c_ret = dds_delete(c_sub);
    dal_ret_t d_ret = RetTypeConverter::GetInstance()->ConvertRetFromCycloneddsToDal(c_ret);
    if (d_ret != DAL_RETCODE_OK) {
        LOG_ERROR_WITH_LOCATION() << "delete subscriber failed." << LOG_ENDL;
        return d_ret;
    }

    auto sub_impl = reinterpret_cast<DalSubscriberImpl*>(sub->handle_);
    delete sub_impl;
    dal_subscriber_free(sub);
    reinterpret_cast<dal_subscriber_t*>(sub)->handle_ = nullptr;

    TopicSertypeManager::GetInstance()->RemoveEntityFromParticipant(c_participant, c_sub);

    return DAL_RETCODE_OK;
}

// DATAREADER
dal_datareader_listener_t*
dal_create_datareader_listener(dal_datareader_listener_callbacks_t callbacks)
{
    dal_datareader_listener_t* listener = dal_datareader_listener_allocate();
    if (listener == nullptr) {
        return nullptr;
    }
    listener->handle_ = (new DataReaderListenerImpl(callbacks));
    if (!listener->handle_) {
        LOG_ERROR() << "DataReaderListenerImpl memory allocation failed." << LOG_ENDL;
        dal_datareader_listener_free(listener);
        return nullptr;
    }

    return listener;
}

dal_ret_t dal_delete_datareader_listener(dal_datareader_listener_t* listener)
{
    if (!listener || !listener->handle_) {
        LOG_ERROR_WITH_LOCATION() << "listener is invalid." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    DataReaderListenerImpl* listener_impl =
        reinterpret_cast<DataReaderListenerImpl*>(listener->handle_);
    if (listener_impl->ref_count_.Get() > 0) {
        LOG_ERROR() << "can't delete datareader listener because it's entity still exist."
                    << LOG_ENDL;
        return DAL_RETCODE_PRECONDITION_NOT_MET;
    }

    delete listener_impl;
    dal_datareader_listener_free(listener);
    reinterpret_cast<dal_datareader_listener_t*>(listener)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

dal_datareader_t* dal_subscriber_create_datareader(dal_subscriber_t* sub,
                                                   dal_topic_description_t* topic_desc,
                                                   dal_datareader_qos_t* qos,
                                                   dal_datareader_listener_t* listener,
                                                   dal_status_mask_t mask)
{
    dal_ret_t d_ret = DAL_RETCODE_OK;

    dds_entity_t c_sub = get_cyclone_subscriber(sub);
    if (c_sub <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone subscriber failed." << LOG_ENDL;
        return nullptr;
    }

    dds_entity_t c_topic = get_cyclone_topic_desc(topic_desc);
    if (c_topic <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone topic description failed." << LOG_ENDL;
        return nullptr;
    }

    dal_datareader_t* d_reader = dal_datareader_allocate();
    if (!d_reader) {
        LOG_ERROR_WITH_LOCATION() << "dal datareader allocate failed." << LOG_ENDL;
        return nullptr;
    }

    DataReaderListenerImpl* c_listener_impl =
        listener ? reinterpret_cast<DataReaderListenerImpl*>(listener->handle_) : nullptr;
    dds_listener_t* c_listener = NULL;
    if (c_listener_impl != nullptr) {
        c_listener = c_listener_impl->CycloneDDSListener();
    }

    dds_qos_t* c_qos = dds_create_qos();
    d_ret = QosConverter::GetInstance()->ConvertDataReaderQosFromDalToCyclonedds(qos, c_qos);
    if (DAL_RETCODE_OK != d_ret) {
        LOG_ERROR_WITH_LOCATION() << "convert datareader qos failed." << LOG_ENDL;
        dal_datareader_free(d_reader);
        dds_delete_qos(c_qos);
        return nullptr;
    }

    dds_entity_t c_reader = dds_create_reader(c_sub, c_topic, c_qos, c_listener);
    dds_delete_qos(c_qos);
    if (c_reader < 0) {
        LOG_ERROR_WITH_LOCATION() << "create datareader failed." << LOG_ENDL;
        dal_datareader_free(d_reader);
        return nullptr;
    }

    d_reader->handle_ = new DalDataReaderImpl(c_reader, c_listener_impl);

    uint32_t c_mask = 0;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToCyclonedds(mask, &c_mask);
    StatusMaskConverter::GetInstance()->FilterDataReaderStatusMask(&c_mask);
    if (dds_set_status_mask(c_reader, c_mask) != DDS_RETCODE_OK) {
        LOG_ERROR_WITH_LOCATION() << "set datareader status mask failed." << LOG_ENDL;
        return d_reader;
    }

    return d_reader;
}

dal_datareader_t* dal_subscriber_create_datareader_with_profile(dal_subscriber_t* sub,
                                                                dal_topic_description_t* topic_desc,
                                                                dal_datareader_qos_t* qos,
                                                                const char* profile_name,
                                                                dal_datareader_listener_t* listener,
                                                                dal_status_mask_t mask)
{
    return dal_subscriber_create_datareader(sub, topic_desc, qos, listener, mask);
}

dal_ret_t dal_subscriber_delete_datareader(dal_subscriber_t* sub, dal_datareader_t* reader)
{
    dds_return_t c_ret;
    dds_entity_t c_sub = get_cyclone_subscriber(sub);
    if (c_sub <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone subscriber failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    dds_entity_t c_reader = get_cyclone_datareader(reader);
    if (c_reader <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone datareader failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (c_sub != dds_get_parent(c_reader)) {
        LOG_ERROR_WITH_LOCATION() << "datareader is not belong to subscriber." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    c_ret = dds_delete(c_reader);
    dal_ret_t d_ret = RetTypeConverter::GetInstance()->ConvertRetFromCycloneddsToDal(c_ret);
    if (d_ret != DAL_RETCODE_OK) {
        LOG_ERROR_WITH_LOCATION() << "delete datareader failed." << c_ret << LOG_ENDL;
        return d_ret;
    }

    auto reader_impl = reinterpret_cast<DalDataReaderImpl*>(reader->handle_);
    delete reader_impl;
    dal_datareader_free(reader);
    reinterpret_cast<dal_datareader_t*>(reader)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

dal_ret_t dal_datareader_read_next_sample(dal_datareader_t* reader,
                                          dal_data data,
                                          dal_sample_info_t* d_sample_info)
{
    const int MAX_SAMPLES = 1;
    dds_sample_info_t c_sample_info;

    dds_entity_t c_reader = get_cyclone_datareader(reader);
    if (c_reader <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone datareader failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (!d_sample_info) {
        LOG_ERROR_WITH_LOCATION() << "sample_info is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    int c_ret = dds_read(c_reader, &data, &c_sample_info, 1, 1);
    SampleInfoConverter::GetInstance()->ConvertSampleInfoFromCycloneddsToDal(d_sample_info,
                                                                             &c_sample_info);
    if (c_ret == 0) {
        return DAL_RETCODE_NO_DATA;
    }
    if (c_ret > 0) {
        return DAL_RETCODE_OK;
    }
    return RetTypeConverter::GetInstance()->ConvertRetFromCycloneddsToDal(c_ret);
}

dal_ret_t dal_datareader_take_next_sample(dal_datareader_t* reader,
                                          dal_data data,
                                          dal_sample_info_t* d_sample_info)
{
    const int MAX_SAMPLES = 1;
    dds_sample_info_t c_sample_info;

    dds_entity_t c_reader = get_cyclone_datareader(reader);
    if (c_reader <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone datareader failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (!d_sample_info) {
        LOG_ERROR_WITH_LOCATION() << "sample_info is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    int c_ret = dds_take(c_reader, &data, &c_sample_info, 1, 1);
    SampleInfoConverter::GetInstance()->ConvertSampleInfoFromCycloneddsToDal(d_sample_info,
                                                                             &c_sample_info);
    if (c_ret == 0) {
        return DAL_RETCODE_NO_DATA;
    }
    if (c_ret > 0) {
        return DAL_RETCODE_OK;
    }
    return RetTypeConverter::GetInstance()->ConvertRetFromCycloneddsToDal(c_ret);
}

dal_guid_t dal_datareader_get_guid(dal_datareader_t* reader)
{
    dal_guid_t d_guid = { };

    dds_entity_t c_reader = get_cyclone_datareader(reader);
    if (c_reader <= 0) {
        LOG_ERROR_WITH_LOCATION() << "get cyclone datareader failed." << LOG_ENDL;
        return d_guid;
    }

    dds_guid_t c_guid;
    if (dds_get_guid(c_reader, &c_guid) != DDS_RETCODE_OK) {
        LOG_ERROR_WITH_LOCATION() << "get gudi by entity failed." << LOG_ENDL;
        return d_guid;
    }
    ddsi_guid_t c_guid_s;
    memcpy(&c_guid_s, &c_guid, sizeof(dds_guid_t));
    GuidConverter::GetInstance()->ConvertGuidFromCycloneddsToDal(&c_guid_s, &d_guid);
    return d_guid;
}
